# socket.rb
# Copyright (C) 2004-2006 Akira TAGOH

# Authors:
#   Akira TAGOH  <at@gclab.org>

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330,
# Boston, MA 02111-1307, USA.

require 'delegate'
require 'singleton'
require 'socket'
require 'thread'
require 'prune/error'
require 'prune/debug'
require 'prune/queue'

Thread.abort_on_exception = true if $DEBUG


module PRUNE

=begin rdoc

== PRUNE::SocketManager

=end

  class SocketManager
    include Singleton
    include PRUNE::Debug

=begin rdoc

=== PRUNE::SocketManager#new

=end

    def initialize
      @readq = []
      @writeq = []
      @errorq = []
      @is_running = false
      @thread = nil
      @threadmap = {}
    end # def initialize

    attr_reader :readq, :writeq, :errorq

=begin rdoc

=== PRUNE::SocketManager#is_running?

=end

    def is_running?
      return @is_running && !@thread.nil?
    end # def is_running?

=begin rdoc

=== PRUNE::SocketManager#start

=end

    def start
      return true if is_running?
      @is_running = true

      @thread = Thread.new do
        while @is_running do
          sleep 0.1
          begin
            retval = IO.select(@readq, @writeq, @errorq, 0)
            unless retval.nil? then
              # for read
              length = retval[0].length
              while length > 0 do
                rs = retval[0][retval[0].length - length]
                if @threadmap.has_key?(rs) && !@threadmap[rs].nil? then
                  # postpone to deal with this to next IO.select
                else
                  @threadmap[rs] = Thread.new(rs) do |rrs|
                    if rrs.class.to_s =~ /Server/ then
                      debug('socket/manager', "Connecting to %s", rrs)
                      s = rrs.accept
                      @readq.push(s)

                      # create an instance for listener
                      if s.respond_to?(:handler) &&
                          s.handler.respond_to?(:emit_by_name) then
                        s.handler.synchronize do |ss|
                        end
                      end
                    else
                      if rrs.eof? then
                        debug('socket/manager', "%s was closed.", rrs)
                        rrs.close
                        @readq.delete(rrs)
                        # destroy an instance for listener
                        if rrs.respond_to?(:handler) &&
                            rrs.handler.respond_to?(:emit_by_name) then
                          begin
                            rrs.handler.synchronize do |srs|
                            end
                          rescue PRUNE::Error::EventServiceNotAvailable
                            # ignore it.
                          end
                        end
                      else
                        str = rrs.gets
                        rrs.inspect =~ /.*host:(.*)\s+port:([0-9]+).*/
                        k = sprintf("%s/%s", $1, $2)
                        k = 'listener' if k == '/'
                        debug('socket/manager', "Received[%s]: %s", k, str.chomp("\r\n"))
                        if rrs.respond_to?(:handler) &&
                            rrs.handler.respond_to?(:emit_by_name) then
                          rrs.handler.synchronize do |srs|
                          end
                        end
                      end
                    end
                    @threadmap.delete(rrs)
                    Thread.exit
                  end
                end
                length -= 1
              end
              # for write
              retval[1].each do |ws|
                FIXME("need to take care of write sockets here.")
                p ws
              end
              # for error
              retval[2].each do |es|
                FIXME("need to take care of error sockets here.")
                p es
              end
            end
          rescue IOError
            # check in the read queue
            @readq.dup.each do |rs|
              if rs.closed? then
                # destroy an instance for listener
                if rs.respond_to?(:handler) &&
                    rs.handler.respond_to?(:emit_by_name) then
                end
                @readq.delete(rs)
              end
            end
            # check in the write queue
            # check in the error queue
          end
        end
        debug('socket/manager', "Stopping...")
        # clean up to finalize.
        @readq.delete_if do |rs|
          debug('socket/manager', "Closing %s", rs)
          rs.stop if rs.respond_to?(:stop)
          rs.close
          true
        end
        @writeq.delete_if do |ws|
          debug('socket/manager', "Closing %s", ws)
          ws.stop if ws.respond_to?(:stop)
          ws.close
          true
        end
        @errorq.delete_if do |es|
          debug('socket/manager', "Closing %s", es)
          es.stop if es.respond_to?(:stop)
          es.close
          true
        end
        debug('socket/manager', 'Stopped')
        Thread.exit
      end
    end # def start

=begin rdoc

=== PRUNE::SocketManager#stop

=end

    def stop
      @is_running = false
      @thread.join
      @thread = nil
    end # def stop

  end # class SocketManager

=begin rdoc

== PRUNE::SocketDelegator

=end

  class SocketDelegator < SimpleDelegator

=begin rdoc

=== PRUNE::SocketDelegator#new(handler, obj)

=end

    def initialize(handler, obj)
      @handler = handler

      super(obj)
    end # def initialize

    attr_reader :handler

    def inspect
      retval = sprintf("#<%s:0x%x %s handler:%s>",
                       self.class, self.object_id, __getobj__.inspect, @handler.inspect)

      return retval
    end # def inspect

  end # class SocketDelegator

=begin rdoc

== PRUNE::ServerSocketDelegator

=end

  class ServerSocketDelegator < PRUNE::SocketDelegator

=begin rdoc

=== PRUNE::ServerSocketDelegator#accept

=end

    def accept
      s = __getobj__.__send__(:accept)
      s.extend(PRUNE::Debug)

      def s.__initvalue__
	@sendq = PRUNE::Queue.new unless defined?(@sendq)
        @threadq = PRUNE::Queue.new unless defined?(@threadq)
        @thread = nil unless defined?(@thread)
        @lasttime = nil unless defined?(@lasttime)
        @stop = false unless defined?(@stop)
	@floodcontrol = true unless defined?(@floodcontrol)
	@status = {} unless defined?(@status)
      end # def __initvalue__
      def s.floodcontrol=(flag)
	@floodcontrol = flag
        unless @floodcontrol then
          self.stop
        end
      end # def floodcontrol=
      def s.status
        __initvalue__
        return @status[Thread.current]
      end # def status
      def s.stop
        __initvalue__
	if @thread.kind_of?(Thread) then
          @stop = true
          until @sendq.empty? do
            @thread.wakeup
            sleep 0.1
          end
          @thread.exit
          @thread = nil
          @stop = false
          @status.clear
        end
      end # def stop
      def s.close
	self.stop
        super
      end # def close
      def s.puts(line)
        __initvalue__
        unless @floodcontrol then
          m = line.chomp("\r\n")
          debug('socket/flood', "Real-sending[no-flood]: %s", m)
          _m = sprintf("%s\r\n", m)
          super(_m)
        else
          raise PRUNE::Error::SocketServiceNotAvailable if @stop == true
          raise self.status unless self.status.nil?
          @sendq.push(line)
          @threadq.push(Thread.current)

          if @thread.nil? then
            @thread = Thread.new do
              Thread.stop
              loop do
                now = Time.now
                flag = (@lasttime.nil? || (now - @lasttime) >= 2) ? true : false
                if flag && !@sendq.empty? then
                  @lasttime = now
                  t = @threadq.shift

                  begin
                    m = @sendq.shift.chomp("\r\n")
                    debug('socket/flood', "Real-sending: %s", m)
                    _m = sprintf("%s\r\n", m)
                    super(_m)
                    @status[t] = nil
                  rescue => e
                    @status[t] = e
                  end
                else
                  if @sendq.empty? then
                    Thread.stop
                  else
                    stime = 1 - (Time.now - @lasttime)
                    sleep stime if stime > 0
                  end
                end
              end # loop
            end # Thread.new
          end
          @thread.wakeup
        end # unless @floodcontrol
      end # def puts

      retval = PRUNE::SocketDelegator.new(@handler, s)

      return retval
    end # def accept

  end # class ServerSocketDelegator

=begin rdoc

== PRUNE::TCPSocket

=end

  class TCPSocket < TCPSocket
    include PRUNE::Debug

=begin rdoc

=== PRUNE::TCPSocket#new(host, port, prefix = nil)

=end

    def initialize(host, port, prefix = nil)
      @host = host
      @port = port
      @prefix = prefix
      @sendq = PRUNE::Queue.new
      @threadq = PRUNE::Queue.new
      @thread = nil
      @lasttime = nil
      @stop = false
      @floodcontrol = true
      @status = {}

      super(host, port)
    end # def initialize

    attr_reader :prefix

    def inspect
      retval = sprintf("#<%s:0x%x host:%s port:%s prefix:%s>",
                       self.class, self.object_id, @host, @port, @prefix)

      return retval
    end # def inspect

=begin rdoc

=== PRUNE::TCPSocket#floodcontrol=

=end

    def floodcontrol=(flag)
      @floodcontrol = flag
      unless @floodcontrol then
        self.stop
      end
    end # def floodcontrol=

=begin rdoc

=== PRUNE::TCPSocket#status

=end

    def status
      return @status[Thread.current]
    end # def status

=begin rdoc

=== PRUNE::TCPSocket#stop

=end

    def stop
      if @thread.kind_of?(Thread) then
        @stop = true
        until @sendq.empty? do
          @thread.wakeup
          sleep 0.1
        end
        @thread.exit
        @thread = nil
        @stop = false
        @status.clear
      end
    end # def stop

=begin rdoc

=== PRUNE::TCPSocket#close

=end

    def close
      self.stop
      super
    end # def close

=begin rdoc

=== PRUNE::TCPSocket#puts(line)

=end

    def puts(line)
      unless @floodcontrol then
        m = line.chomp("\r\n")
        debug('socket/flood', "Real-sending[no-flood]: %s", m)
        _m = sprintf("%s\r\n", m)
        super(_m)
      else
        raise PRUNE::Error::SocketServiceNotAvailable if @stop == true
        raise self.status unless self.status.nil?
        @sendq.push(line)
        @threadq.push(Thread.current)

        if @thread.nil? then
          @thread = Thread.new do
            Thread.stop
            loop do
              now = Time.now
              flag = (@lasttime.nil? || (now - @lasttime) >= 2) ? true : false
              if flag && !@sendq.empty? then
                @lasttime = now
                t = @threadq.shift

                begin
                  m = @sendq.shift.chomp("\r\n")
                  debug('socket/flood', "Real-sending: %s", m)
                  _m = sprintf("%s\r\n", m)
                  super(_m)
                  @status[t] = nil
                rescue => e
                  @status[t] = e
                end
              else
                if @sendq.empty? then
                  Thread.stop
                else
                  stime = 1 - (Time.now - @lasttime)
                  sleep stime if stime > 0
                end
              end
            end # loop
          end # Thread.new
        end
        @thread.wakeup
      end # unless @floodcontrol
    end # def puts

  end # class TCPSocket

=begin rdoc

== PRUNE::TCPServer

=end

  class TCPServer < TCPServer

=begin rdoc

=== PRUNE::TCPServer#new(host, port, prefix = nil)

=end

    def initialize(host, port, prefix = nil)
      @host = host
      @port = port
      @prefix = prefix

      super(host, port)
    end # def initialize

    def inspect
      retval = sprintf("#<%s:0x%x host:%s port:%s prefix:%s>",
                       self.class, self.object_id, @host, @port, @prefix)

      return retval
    end # def inspect

  end # class TCPServer

end # module PRUNE
